package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/13
 */
public class App03_DwdMhsPrescChina extends BaseSqlApp {
    public static void main(String[] args) {
        new App03_DwdMhsPrescChina().init(
                5003,
                "App3_DwdMhsPrescChina",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target(" +
                "RECIPEL_ID                       STRING," +
                "HOSPITAL_NAME                    STRING," +
                "HOSPITAL_CODE                    STRING," +
                "AGE_UNIT                         STRING," +
                "AGE                              STRING," +
                "SEX_NAME                         STRING," +
                "SEX_CODE                         STRING," +
                "ID_CARD_NUMBER                   STRING," +
                "PATIENT_VISIT_CATEGORY_CODE      STRING," +
                "PATIENT_VISIT_CATEGORY           STRING," +
                "OUTPATIENT_VISIT_FLOW_ID         STRING," +
                "OUTPATIENT_NO                    STRING," +
                "PATIENT_ID                       STRING," +
                "PATIENT_NAME                     STRING," +
                "DOMAIN_ID                        STRING," +
                "VISIT_TIMES                      STRING," +
                "OPEN_DEPT_CODE                   STRING," +
                "OPEN_DEPT                        STRING," +
                "AUTHOR_CODE                      STRING," +
                "AUTHOR_NAME                      STRING," +
                "CUSTODIAN_CODE                   STRING," +
                "CUSTODIAN_NAME                   STRING," +
                "AUDIT_SIGNDATE                   STRING," +
                "AUDIT_CODE                       STRING," +
                "AUDIT_NAME                       STRING," +
                "DEPLOY_SIGNDATE                  STRING," +
                "DEPLOY_CODE                      STRING," +
                "DEPLOY_NAME                      STRING," +
                "CHECK_SIGNDATE                   STRING," +
                "CHECK_CODE                       STRING," +
                "CHECK_NAME                       STRING," +
                "DISPENSER_SIGNDATE               STRING," +
                "DISPENSER_CODE                   STRING," +
                "DISPENSER_NAME                   STRING," +
                "RECIPEL_DAYS                     STRING," +
                "PDRUG_GROUP                      STRING," +
                "TCM_PIECES                       STRING," +
                "TCM_NUM_TABLETS                  STRING," +
                "TCM_METHOD                       STRING," +
                "CHINA_USED_METHOD                STRING," +
                "PCATEGORY_CODE                   STRING," +
                "PCATEGORY_NAME                   STRING," +
                "THERAPEUTIC_PRINCIPLE            STRING," +
                "PDRUGS_FEE                       STRING," +
                "RECIPEL_REMARK                   STRING," +
                "VISIT_DATE                       STRING," +
                "RECIPE_FLAG                      STRING," +
                "DIAG_CODE                        STRING," +
                "DIAG_NAME                        STRING," +
                "CHINA_DISEASE_CODE               STRING," +
                "CHINA_DISEASE_NAME               STRING," +
                "CHINA_SYNDROME_CODE              STRING," +
                "CHINA_SYNDROME_NAME              STRING," +
                "VISIT_ID                         STRING," +
                "primary key (RECIPEL_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_PRESC));
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id as RECIPEL_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "visit_times as VISIT_TIMES , " +
                "open_dept_code as OPEN_DEPT_CODE , " +
                "open_dept as OPEN_DEPT , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "audit_signdate as AUDIT_SIGNDATE , " +
                "audit_code as AUDIT_CODE , " +
                "audit_name as AUDIT_NAME , " +
                "deploy_signdate as DEPLOY_SIGNDATE , " +
                "deploy_code as DEPLOY_CODE , " +
                "deploy_name as DEPLOY_NAME , " +
                "check_signdate as CHECK_SIGNDATE , " +
                "check_code as CHECK_CODE , " +
                "check_name as CHECK_NAME , " +
                "dispenser_signdate as DISPENSER_SIGNDATE , " +
                "dispenser_code as DISPENSER_CODE , " +
                "dispenser_name as DISPENSER_NAME , " +
                "recipel_days as RECIPEL_DAYS , " +
                "pdrug_group as PDRUG_GROUP , " +
                "tcm_pieces as TCM_PIECES , " +
                "tcm_num_tablets as TCM_NUM_TABLETS , " +
                "tcm_method as TCM_METHOD , " +
                "tcm_drug_use as CHINA_USED_METHOD , " +
                "pcategory_code as PCATEGORY_CODE , " +
                "pcategory_name as PCATEGORY_NAME , " +
                "principle_and_method as THERAPEUTIC_PRINCIPLE , " +
                "pdrugs_fee as PDRUGS_FEE , " +
                "recipel_remark as RECIPEL_REMARK , " +
                "visit_date as VISIT_DATE , " +
                "'1' as RECIPE_FLAG , " +
                "case when diag_type='西医诊断' then diag_code else '' end as DIAG_CODE , " +
                "case when diag_type='西医诊断' then diag_name else '' end as DIAG_NAME , " +
                "case when diag_type='中医病名' then diag_code else '' end as CHINA_DISEASE_CODE , " +
                "case when diag_type='中医病名' then diag_name else '' end as CHINA_DISEASE_NAME , " +
                "case when diag_type='中医证候' then diag_code else '' end as CHINA_SYNDROME_CODE , " +
                "case when diag_type='中医证候' then diag_name else '' end as CHINA_SYNDROME_NAME ," +
                "visit_id as VISIT_ID  " +
                "from hdsd00_04_02 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    ");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_04_02
        tableEnv.executeSql("create table hdsd00_04_02 (" +
                "pk                        string," +
                "upload_time               string," +
                "status                    string," +
                "empi                      string," +
                "encounter_id              string," +
                "patient_id                string," +
                "patient_domain            string," +
                "visit_domain              string," +
                "visit_id                  string," +
                "visit_times               string," +
                "unique_id                 string," +
                "xds_type                  string," +
                "xds_name                  string," +
                "effective_time            string," +
                "xds_version               string," +
                "out_patient_id            string," +
                "domain_id                 string," +
                "recipel_id                string," +
                "patient_type              string," +
                "identify_no               string," +
                "patient_typecode          string," +
                "patient_name              string," +
                "sex                       string," +
                "sex_code                  string," +
                "age                       string," +
                "age_unit                  string," +
                "open_dept_code            string," +
                "open_dept                 string," +
                "hospital_code             string," +
                "hospital_name             string," +
                "write_time                string," +
                "author_code               string," +
                "author_name               string," +
                "custodian_code            string," +
                "custodian_name            string," +
                "audit_signdate            string," +
                "audit_code                string," +
                "audit_name                string," +
                "check_signdate            string," +
                "deploy_code               string," +
                "deploy_name               string," +
                "deploy_signdate           string," +
                "check_code                string," +
                "check_name                string," +
                "dispenser_signdate        string," +
                "dispenser_code            string," +
                "dispenser_name            string," +
                "recipel_days              string," +
                "pdrug_group               string," +
                "tcm_pieces                string," +
                "tcm_num_tablets           string," +
                "tcm_method                string," +
                "tcm_drug_use              string," +
                "pcategory_code            string," +
                "pcategory_name            string," +
                "pdrugs_fee                string," +
                "recipel_remark            string," +
                "principle_and_method      string," +
                "visit_date                string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_04_02","App3_DwdMhsPrescChina"));
        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App3_DwdMhsPrescChina");
    }
}
